//
// Created by 62660 on 2022-02-25.
//

#include "ThreadPoll.h"

ThreadPoll::ThreadPoll(int size) : stop(false) {
    for (int i = 0; i < size; ++i) {
        threadsVec.emplace_back(
                std::thread([this]() {
                    while (true) {
                        std::function<void()> task;
                        {
                            //在这个{}作用域内对std::mutex加锁，出了作用域会自动解锁，不需要调用unlock()
                            std::unique_lock<std::mutex> lock(tasks_mtx);
                            cond_variable.wait(lock, [this]() {
                                return stop || !tasksQueue.empty();
                            });
                            task = tasksQueue.front();
                            tasksQueue.pop();
                        }
                        task();
                    }
                })
        );
    }
}

ThreadPoll::~ThreadPoll() {
    {
        std::unique_lock<std::mutex> lock(tasks_mtx);
        stop = true;
    }
    cond_variable.notify_all();
    for (std::thread &th : threadsVec) {
        if(th.joinable())
            th.join();
    }
}

void ThreadPoll::add(std::function<void()> func) {
    //在这个{}作用域内对std::mutex加锁，出了作用域会自动解锁，不需要调用unlock()
    {
        std::unique_lock<std::mutex> lock(tasks_mtx);
        if (stop) {
            throw std::runtime_error("ThreadPoll already stop, can't add task any more");
        }
        tasksQueue.emplace(func);
    }
    cond_variable.notify_one(); //通知一次条件变量
}
